package com.shengzai.flink.state

import com.alibaba.fastjson2.{JSON, JSONObject}
import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.lang
import java.util.NoSuchElementException
import scala.collection.mutable
import scala.collection.mutable.ListBuffer

object Demo1CarAndAvgSpeed {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    val socketDS: DataStream[String] = env.socketTextStream("master", 8888)

    val mapDS: DataStream[(String, Double)] = socketDS.map(
      line => {
        val jsonObj: JSONObject = JSON.parseObject(line)
        val road: String = jsonObj.getString("road_id")
        val speed: Double = jsonObj.getDouble("speed")
        (road, speed)
      }
    )
    val keyByDS: KeyedStream[(String, Double), String] = mapDS.keyBy(_._1)

    keyByDS.process(new KeyedProcessFunction[String,(String,Double),(String,Double)] {
      //同一个Task共享变量
      val hashMap = new mutable.HashMap[String, ListBuffer[Double]]()

      override def processElement(value: (String, Double),
                                  ctx: KeyedProcessFunction[String, (String, Double), (String, Double)]#Context,
                                  out: Collector[(String, Double)]): Unit = {
        val (road:String,spend:Double) =value

        var spends: Double = hashMap.getOrElse(road, ListBuffer[Double](0,0)).head

        var count: Int = hashMap.getOrElse(road, ListBuffer[Double](0, 0)).last.toInt

        spends =spend+spends

        count =count+1
        hashMap.put(road,ListBuffer[Double](spends,count))

        val avgSpeed: Double = spends / count


        out.collect(road,avgSpeed)

      }
    }).print()

    env.execute()


//    var count :Int = 0
//    var tmpSpeed :Double =0
//    val hashMap = new mutable.HashMap[String, Double]()
//    mapDS.process(new ProcessFunction[(String,Double),(String,Double)] {
//      override def processElement(value: (String, Double),
//                                  ctx: ProcessFunction[(String, Double), (String, Double)]#Context,
//                                  out: Collector[(String, Double)]): Unit = {
//        val (road:String, speed:Double) =value
//
//
//
//       // val kaKouSpeed: Double = hashMap.getOrElse(road, 0)
//       // hashMap.put(road,speed)
//        tmpSpeed= tmpSpeed+speed
//        count =count+1


      }





}
